package avicit.bdp.dcs.threshold.concurrency;

import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.threshold.utils.TestUtils;
import avicit.platform6.core.rest.msg.ResponseMsg;
import com.alibaba.fastjson2.JSON;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.Assert.assertEquals;

/** 并发测试; */
public class ConcurrencyTest {
  private RestTemplate restTemplate = new RestTemplate();

  private static final int BATCH_SIZE = 10;

  private static String srcTble = "concurrency_src_" + System.currentTimeMillis();

  private String[] dstTblNames = new String[BATCH_SIZE];

  private AtomicInteger taskExeResult = new AtomicInteger(0);

  /** 所有用例执行前准备工作 */
  @BeforeClass
  public static void init() {
    // TODO:
  }

  /** 单个测试用例运行前准备工作 */
  @Before
  public void setUp() {
    TestUtils.insert2Mysql(srcTble);
  }

  /** 单个测试用例运行后销毁工作 */
  @After
  public void tearDown() {
    TestUtils.deleteFromMysql(srcTble);
    for (String tblName : dstTblNames) {
      TestUtils.deleteFromMysql(tblName);
    }
  }

  /** 所有用例执行完成后准备工作 */
  @AfterClass
  public static void destory() {
    // TODO:
  }

  /** 新建Job，返回创建的JobId */
  private String createJob(String dstTble) {
    String jobId = null;
    try {
      // step1:保存数据源
      ResponseEntity<ResponseMsg> result =
          restTemplate.postForEntity(
              TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/saveDataSource",
              TestUtils.buildSaveDataSourceParas(
                  2,
                  "并发测试门槛用例",
                  TestUtils.MYSQL_SRC_DATASOURCE_ID,
                  TestUtils.MYSQL_DST_DATASOURCE_ID,
                  1),
              ResponseMsg.class);
      TestUtils.handleFailedRequest(result);
      if (!result.getBody().getRetCode().equals("200")) {
        throw new IOException("保存数据源信息失败");
      }

      // 获取创建的JobId
      ResponseMsg responseMsg = result.getBody();
      Map<String, String> jobMap = (Map<String, String>) responseMsg.getResponseBody();
      jobId = jobMap.get("jobId");

      // step2:保存表映射关系
      result =
          restTemplate.postForEntity(
              TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/saveTableMapping",
              TestUtils.buildSaveTblMapping(2, jobId, srcTble, dstTble),
              ResponseMsg.class);
      TestUtils.handleFailedRequest(result);
      if (!result.getBody().getRetCode().equals("200")) {
        throw new IOException("保存表映射关心失败");
      }

      // step3:任务配置检查
      result =
          restTemplate.postForEntity(
              TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/checkTaskConfig",
              TestUtils.buildCheckTaskConfig(jobId),
              ResponseMsg.class);
      TestUtils.handleFailedRequest(result);
      if (!result.getBody().getRetCode().equals("200")) {
        throw new IOException("任务配置检查失败");
      }

      // step4:预检查
      result =
          restTemplate.postForEntity(
              TestUtils.HOSTPORT + "/api/bdp/dcs/precheck/PrecheckRest" + "/executePrecheck",
              TestUtils.buildPreChecks(jobId),
              ResponseMsg.class);
      TestUtils.handleFailedRequest(result);
      if (!result.getBody().getRetCode().equals("200")) {
        throw new IOException("预检查失败");
      }

      // step5:更新Job状态
      result =
          restTemplate.postForEntity(
              TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/updateJobStatus",
              TestUtils.buildUpdateJobStatus(jobId),
              ResponseMsg.class);
      TestUtils.handleFailedRequest(result);
      if (!result.getBody().getRetCode().equals("200")) {
        throw new IOException("更新Job状态失败");
      }
    } catch (Exception e) {
      jobId = null;
    }

    return jobId;
  }

  /**
   * 并发创建jobs；
   *
   * @return
   */
  private String[] concurrencyCreateJobs() {
    String[] jobIds = new String[BATCH_SIZE];
    CountDownLatch downLatch = new CountDownLatch(BATCH_SIZE);

    for (int i = 0; i < BATCH_SIZE; i++) {
      final int index = i;
      new Thread(
              new Runnable() {
                @Override
                public void run() {
                  try {
                    System.out.println("开始执行创建任务");
                    Thread.sleep(100);
                    long currTime = System.currentTimeMillis();
                    String jobId = createJob("ConcurrencyTest_" + currTime + "_" + index);
                    if (jobId == null) {
                      System.out.println(
                          String.format("创建任务失败，dstTbl=%s", "ConcurrencyTest_" + index));
                    } else {
                      jobIds[index] = jobId;
                      taskExeResult.incrementAndGet();
                    }

                    dstTblNames[index] = "ConcurrencyTest_" + currTime + "_" + index;
                    System.out.println("创建任务执行完成");

                    // 当前线程执行完毕后，计数值-1
                    downLatch.countDown();
                  } catch (InterruptedException e) {
                    e.printStackTrace();
                  }
                }
              })
          .start();
    }

    try {
      System.out.println("等待所有创建Job子任务执行完成...");
      downLatch.await();
      System.out.println("所有创建Job子任务执行完成");

      // 判断所有任务是否执行成功
      System.out.println("创建任务，taskExeResult=" + taskExeResult.get() + ",预期数量=" + BATCH_SIZE);
      assertEquals(taskExeResult.get(), BATCH_SIZE);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    // 任务执行结果初始化
    taskExeResult.set(0);

    return jobIds;
  }

  /** 并发启动采集任务 */
  private void concurrencyStartJobs(String[] jobIds) {
    if (jobIds == null || jobIds.length <= 0) {
      System.out.println("数组为null");
      return;
    }

    CountDownLatch downLatch = new CountDownLatch(BATCH_SIZE);
    for (int i = 0; i < BATCH_SIZE; i++) {
      String currJobId = jobIds[i];
      new Thread(
              new Runnable() {
                @Override
                public void run() {
                  try {
                    System.out.println("开始执行启动任务,jobId=" + currJobId);
                    Thread.sleep(100);
                    ResponseEntity<ResponseMsg> result =
                        restTemplate.postForEntity(
                            TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/startJobByIds",
                            TestUtils.buildStartJob(currJobId),
                            ResponseMsg.class);
                    TestUtils.handleFailedRequest(result);
                    if (!result.getBody().getRetCode().equals("200")) {
                      System.out.println("执行启动任务失败,jobId=" + currJobId);
                    } else {
                      taskExeResult.incrementAndGet();
                    }
                    System.out.println("启动jobId=" + currJobId + "任务执行完成");

                    // 当前线程执行完毕后，计数值-1
                    downLatch.countDown();
                  } catch (InterruptedException e) {
                    e.printStackTrace();
                  }
                }
              })
          .start();
    }

    try {
      System.out.println("等待所有启动Job子任务执行完成...");
      downLatch.await();
      System.out.println("所有启动Job子任务执行完成");

      // 判断所有任务是否执行成功
      System.out.println("启动任务，taskExeResult=" + taskExeResult.get() + ",预期数量=" + BATCH_SIZE);
      assertEquals(taskExeResult.get(), BATCH_SIZE);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    // 任务执行结果初始化
    taskExeResult.set(0);
  }

  /** 并发删除采集任务 */
  private void concurrencyDeleteJobs(String[] jobIds) {
    if (jobIds == null || jobIds.length <= 0) {
      System.out.println("数组为null");
      return;
    }

    CountDownLatch downLatch = new CountDownLatch(BATCH_SIZE);
    for (int i = 0; i < BATCH_SIZE; i++) {
      String currJobId = jobIds[i];
      new Thread(
              new Runnable() {
                @Override
                public void run() {
                  try {
                    System.out.println("开始执行删除任务,jobId=" + currJobId);
                    Thread.sleep(100);
                    ResponseEntity<ResponseMsg> result =
                        restTemplate.exchange(
                            TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/deleteJobByIds",
                            HttpMethod.DELETE,
                            TestUtils.buildDeleteJob(currJobId),
                            ResponseMsg.class);
                    TestUtils.handleFailedRequest(result);
                    if (!result.getBody().getRetCode().equals("200")) {
                      System.out.println("执行删除任务失败,jobId=" + currJobId);
                    } else {
                      taskExeResult.incrementAndGet();
                    }
                    System.out.println("删除jobId=" + currJobId + "任务执行完成");

                    // 当前线程执行完毕后，计数值-1
                    downLatch.countDown();
                  } catch (InterruptedException e) {
                    e.printStackTrace();
                  }
                }
              })
          .start();
    }

    try {
      System.out.println("等待所有删除Job子任务执行完成...");
      downLatch.await();
      System.out.println("所有删除Job子任务执行完成");

      // 判断所有任务是否执行成功
      System.out.println("删除任务，taskExeResult=" + taskExeResult.get() + ",预期数量=" + BATCH_SIZE);
      assertEquals(taskExeResult.get(), BATCH_SIZE);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    // 任务执行结果初始化
    taskExeResult.set(0);
  }

  /** 并发停止采集任务 */
  private void concurrencyFinishJobs(String[] jobIds) {
    if (jobIds == null || jobIds.length <= 0) {
      System.out.println("数组为null");
      return;
    }

    CountDownLatch downLatch = new CountDownLatch(BATCH_SIZE);
    for (int i = 0; i < BATCH_SIZE; i++) {
      String currJobId = jobIds[i];
      new Thread(
              new Runnable() {
                @Override
                public void run() {
                  try {
                    System.out.println("开始执行停止任务,jobId=" + currJobId);
                    Thread.sleep(100);
                    ResponseEntity<ResponseMsg> result =
                        restTemplate.postForEntity(
                            TestUtils.HOSTPORT + "/api/bdp/dcs" + "/job/JobRest/finishJobByIds",
                            TestUtils.buildStartJob(currJobId),
                            ResponseMsg.class);
                    TestUtils.handleFailedRequest(result);
                    if (!result.getBody().getRetCode().equals("200")) {
                      System.out.println("执行停止任务失败,jobId=" + currJobId);
                      taskExeResult.incrementAndGet();
                    } else {
                      // TODO：任务停止无意义，后台并非真正停止
                      taskExeResult.incrementAndGet();
                    }
                    System.out.println("停止jobId=" + currJobId + "任务执行完成");

                    // 当前线程执行完毕后，计数值-1
                    downLatch.countDown();
                  } catch (InterruptedException e) {
                    e.printStackTrace();
                  }
                }
              })
          .start();
    }

    try {
      System.out.println("等待所有停止Job子任务执行完成...");
      downLatch.await();
      System.out.println("所有停止Job子任务执行完成");

      // 判断所有任务是否执行成功
      System.out.println("停止任务，taskExeResult=" + taskExeResult.get() + ",预期数量=" + BATCH_SIZE);
      assertEquals(taskExeResult.get(), BATCH_SIZE);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    // 任务执行结果初始化
    taskExeResult.set(0);
  }

  /***********************************************Begin 测试用例 ***********************************************/
  /**
   * step1:并发创建任务；step2:并发启动采集任务：step3:判断采集任务状态;(预期结果：success,success,success)
   *
   * @throws Exception
   */
  @Test(timeout = 5 * 60 * 1000)
  public void testBatchExecutorTest1() throws Exception {
    String[] jobIds = new String[0];

    try {
      // step1:；并发创建任务
      jobIds = concurrencyCreateJobs();

      // step2:并发启动任务
      concurrencyStartJobs(jobIds);

      // step3:查看任务执行状态
      Thread.sleep(60 * 1000);
      ResponseEntity<ResponseMsg> result =
          restTemplate.exchange(
              TestUtils.HOSTPORT
                  + "/api/bdp/dcs/job/JobRest"
                  + "/getJobList"
                  + "?jobStatus&sort&name="
                  + "并发测试门槛用例"
                  + "&jobType=1&currentPage=1&pageSize=10",
              HttpMethod.GET,
              TestUtils.buildGetHttpEntity(),
              ResponseMsg.class);
      TestUtils.handleFailedRequest(result);
      ResponseMsg body = result.getBody();
      LinkedHashMap<String, Object> map = (LinkedHashMap<String, Object>) body.getResponseBody();
      List<Object> finalBody = (List<Object>) map.get("result");
      for (Object obj : finalBody) {
        LinkedHashMap<String, String> jobObj = (LinkedHashMap<String, String>) obj;
        JobDTO jobInfo = JSON.parseObject(JSON.toJSONString(jobObj), JobDTO.class);
        if (Arrays.asList(jobIds).contains(jobInfo.getId())) {
          System.out.println(
              String.format("任务id:%s,任务状态:%s", jobInfo.getId(), jobInfo.getJobStatus()));

          // 判断任务是否同步完成;
          assertEquals(7 == jobInfo.getJobStatus(), true);

          // 判断同步速度;（备注：源表中只有一行数据，大小固定）
          assertEquals(jobInfo.getReadSucceedBytes() == 18, true);

          break;
        }
      }

    } finally {
      // step4；并发删除采集任务
      concurrencyDeleteJobs(jobIds);
    }
  }

  /** step1:并发创建任务；step2:并发启动采集任务：step3:并发停止采集任务;(预期结果：success,success,fail) */
  @Test(timeout = 5 * 60 * 1000)
  public void testBatchExecutorTest2() throws Exception {
    String[] jobIds = new String[0];

    try {
      // step1:；并发创建任务
      jobIds = concurrencyCreateJobs();

      // step2:并发启动任务
      concurrencyStartJobs(jobIds);

      // step3:并发停止任务
      // 延时10s，等待任务全部启动完成
      Thread.sleep(3 * 1000);
      concurrencyFinishJobs(jobIds);
    } finally {
      // step4；并发删除采集任务
      concurrencyDeleteJobs(jobIds);
    }
  }
}
